/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package es

import (
	"strings"
	"time"

	"hundsun.com/hsl/hschain/common/cache"
	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/third/elasticsearch"
	"hundsun.com/hsl/hschain/common/types"
	drivers "hundsun.com/hsl/hschain/store/db"
)

const (
	// DbName ...
	DbName = "es"
)

// Service es service
type Service struct {
	EsClient *elasticsearch.Client
}

// BatchService batch es service
type BatchService struct {
	esClient *elasticsearch.Client
	cache    *cache.Cache
}

var log logging.Log

func init() {
	drivers.Register(DbName, New)

	log = logging.GetLogWithCommonConf(logging.ConfigKeyElasticSearchLogLevel, logging.LogFieldStore, logging.LogFieldStore).WithField(logging.LogFieldStore, "esSrv")
}

// New new Service
func New(cfg *config.PluginConfig) (drivers.Database, error) {
	esService := &Service{}
	c, err := elasticsearch.Map2EsConfig(cfg.Parameters)
	if err != nil {
		return nil, err
	}
	esService.EsClient, err = elasticsearch.NewClient(c)
	if err != nil {
		return nil, err
	}
	return esService, nil
}

func splitIndexAndKey(key string) (string, string, error) {
	if key == "" {
		return "", "", types.ErrEmptyDocID
	}

	s := strings.SplitN(key, drivers.PrefixSeparator, 2)
	if len(s) < 2 {
		return drivers.DefaultIndexName, key, nil
	}
	return s[0], s[1], nil
}

// Set set kv
func (s *Service) Set(key []byte, value []byte) error {
	index, docID, err := splitIndexAndKey(string(key))
	if err != nil {
		return err
	}

	defer log.ElapsedTime("Set", time.Now())

	return s.EsClient.Set(index, docID, value)
}

// Has ...
func (s *Service) Has(key []byte) (bool, error) {
	index, docID, err := splitIndexAndKey(string(key))
	if err != nil {
		return false, err
	}

	defer log.ElapsedTime("Has", time.Now())

	return s.EsClient.Has(index, docID)
}

// Delete delete key
func (s *Service) Delete(key []byte) error {
	index, docID, err := splitIndexAndKey(string(key))
	if err != nil {
		return err
	}

	defer log.ElapsedTime("Delete", time.Now())

	return s.EsClient.Delete(index, docID)
}

// Close ...
func (s *Service) Close() {
	// no need to implement
}

// Get get key
func (s *Service) Get(key []byte) ([]byte, error) {
	index, docID, err := splitIndexAndKey(string(key))
	if err != nil {
		return nil, err
	}

	defer log.ElapsedTime("Get", time.Now())

	return s.EsClient.Get(index, docID)
}

// Gets 批量查询 返回数组允许元素为nil
func (s *Service) Gets(keys [][]byte) ([][]byte, error) {
	if len(keys) == 0 {
		return nil, types.ErrNilRequests
	}

	requests := make([]elasticsearch.GetRequest, len(keys))
	for i, key := range keys {
		index, docID, err := splitIndexAndKey(string(key))
		if err != nil {
			return nil, err
		}
		requests[i] = elasticsearch.GetRequest{Index: index, DocID: docID}
	}

	defer log.ElapsedTime("Gets", time.Now())

	return s.EsClient.Gets(requests)
}

// NewBatch 批量操作接口（使用本地缓存）, 忽略sync入参
func (s *Service) NewBatch(sync bool) drivers.Batch {
	return &BatchService{s.EsClient, cache.New()}
}

// NewIteratorWithRange new range iterator
func (s *Service) NewIteratorWithRange(start []byte, limit []byte) drivers.Iterator {
	// TODO es的迭代器暂时没有实现
	panic("es not support iterator")
	return nil
}

// NewIteratorWithPrefix new prefix iterator
func (s *Service) NewIteratorWithPrefix(prefix []byte) drivers.Iterator {
	// TODO es的迭代器暂时没有实现
	panic("es not support iterator")
	return nil
}

type esList struct {
	*Service
}

// NewList new es list
func (s *Service) NewList() drivers.Lister {
	return &esList{s}
}

// ListRange count < 0 则对取值数量不做限制
func (esl *esList) ListRange(start, limit []byte, count int32, opt drivers.Option) ([][]byte, [][]byte, error) {
	// TODO es的迭代器暂时没有实现
	panic("es not support list")
	return nil, nil, nil
}

// ListPrefix count < 0 则对取值数量不做限制
func (esl *esList) ListPrefix(prefix []byte, count int32, opt drivers.Option) ([][]byte, [][]byte, error) {
	// TODO es的迭代器暂时没有实现
	panic("es not support list")
	return nil, nil, nil
}

// RangeCount 查询Range范围内迭代个数
func (esl *esList) RangeCount(start, limit []byte) int64 {
	// TODO es的迭代器暂时没有实现
	panic("es not support list")
	return -1
}

// PrefixCount 查询prefix迭代个数
func (esl *esList) PrefixCount(prefix []byte) int64 {
	// TODO es的迭代器暂时没有实现
	panic("es not support list")
	return -1
}

// ValueSize 缓存的记录条数
func (s *BatchService) ValueSize() int {
	return s.cache.Size()
}

// Write 将记录从缓存写到es（写入完成后，注意手动调用Reset清空缓存）
func (s *BatchService) Write() error {
	var requests []elasticsearch.SetRequest
	for k, v := range s.cache.Items() {
		index, docID, _ := splitIndexAndKey(k)
		requests = append(requests, elasticsearch.SetRequest{Index: index, DocID: docID, Value: v.Object.([]byte)})
	}

	return s.esClient.Sets(requests)
}

// Reset 清空缓存
func (s *BatchService) Reset() {
	s.cache.Clear()
}

// Set 将记录保存到缓存（key存在则覆盖）
func (s *BatchService) Set(key []byte, value []byte) error {
	if _, _, err := splitIndexAndKey(string(key)); err != nil {
		return err
	}
	s.cache.Set(string(key), value)
	return nil
}

// Delete 从缓存删除记录
func (s *BatchService) Delete(key []byte) error {
	s.cache.Delete(string(key))
	return nil
}
